/******************************************************************************
* 	Copyright (c) 2007-2008 All rights reserved
*		Asemantics S.r.l
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions
* are met:
*
* 1. Redistributions of source code must retain the above copyright
*    notice, this list of conditions and the following disclaimer. 
*
* 2. Redistributions in binary form must reproduce the above copyright
*    notice, this list of conditions and the following disclaimer in
*    the documentation and/or other materials provided with the
*    distribution.
*
* 3. The end-user documentation included with the redistribution,
*    if any, must include the following acknowledgment:
*       "This product includes software developed by 
*	 Asemantics S.r.l."
*    Alternately, this acknowledgment may appear in the software itself,
*    if and wherever such third-party acknowledgments normally appear.
*
* 4. All advertising materials mentioning features or use of this software
*    must display the following acknowledgement:
*    This product includes software developed by Asemantics S.r.l.
*    the Semantic Web company, Rome, London, Leiden and its contributors. 
*
* 5. Neither the name of the company nor the names of its contributors
*    may be used to endorse or promote products derived from this software
*    without specific prior written permission.
*
* THIS SOFTWARE IS PROVIDED BY THE REGENTS AND CONTRIBUTORS ``AS IS'' AND
* ANY EXPRESSED OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
* PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE REGENTS OR CONTRIBUTORS BE
* LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
* CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
* SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
* INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
* CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED
* OF THE POSSIBILITY OF SUCH DAMAGE.
*
* Author: Andrea Marchesini - baku@asemantics.com
*
******************************************************************************/

#ifndef G_OS_WIN32
#  include <sys/types.h>
#  include <signal.h>
#endif

#include <nxml.h>

#include <controller.h>
#include <configure.h>
#include <iostring.h>
#include <process.h>
#include <istance.h>
#include <log.h>

typedef enum
{
  ISTANCE_IO_IN,
  ISTANCE_IO_OUT,
  ISTANCE_IO_ERR
} TBIstanceIO;

static void istance_wait_free (TBIstanceWait * wait);

static gchar **istance_command_inline (gchar ** argv, TBIOString * input);

static void istance_env (GList * env);
static gchar **istance_env_command (gchar ** argv, GList * env);

static gboolean istance_is_to_propage (TBIstance * istance, guint order);
static void istance_propage_check (TBIstance * istance, TBProcess * process,
				   guint order);

static void istance_check_wait (TBGlobal * data);

#ifdef G_OS_UNIX
static gboolean istance_spawn (gchar ** argv,
			       GSpawnChildSetupFunc child_setup,
			       gpointer user_data, GPid * pid,
			       gint * standard_input, gint * standard_output,
			       gint * standard_error, GError ** error);
#endif

static void istance_io_close (TBIstance * istance, TBIstanceIO io);

/* NEW *********************************************************************/
static gchar **istance_split (gchar * ptr);
static gboolean istance_io_out (GIOChannel * source, GIOCondition condition,
				TBIstance * istance);
static gboolean istance_io_in (GIOChannel * source, GIOCondition condition,
			       TBIstance * istance);
static gboolean istance_io_in_err (GIOChannel * source,
				   GIOCondition condition,
				   TBIstance * istance);

/* This function execs a process and creates a istances: */
static void
istance_new_exec (TBGlobal * data, TBProcess * process, guint order,
		  GList * parents, GList * metaprocesses,
		  TBIOString * input_buffer, gboolean force)
{
  TBIstance *istance;
  GError *error = NULL;
  gchar **argv, **tmp;
  gint in, out, err;
  GPid pid;

#ifdef TB_DEBUG
  debug (__func__, __LINE__,
	 "process: %s, order: %d, parents: %p, metaprocess: %p, input: %p, force: %s",
	 process->id, order, parents, metaprocesses, input_buffer,
	 force == TRUE ? "true" : "false");
#endif

  /* If this project has too many istances running, I save the current one */
  if (force == FALSE
      &&
      ((process->istances_wait_numb != 0
	|| (process->istances != 0
	    && process->istances_running >= process->istances))))
    {
      TBIstanceWait *wait;

      wait = g_malloc0 (sizeof (TBIstanceWait));

      wait->input = input_buffer;
      io_string_ref (input_buffer);

      wait->process = process;
      wait->process_order = order;

      wait->parents = g_list_copy (parents);
      g_list_foreach (wait->parents, (GFunc) istance_ref, NULL);

      /* This part has to be locked: */
      g_mutex_lock (data->mutex);

      /* Duplicate the metaprocess for this waiting istance: */
      if (metaprocesses)
	{
	  wait->metaprocesses = g_list_copy (metaprocesses);
	  g_list_foreach (wait->metaprocesses, (GFunc) process_ref, NULL);
	}

      data->istances_wait = g_list_append (data->istances_wait, wait);
      data->istances_wait_numb++;
      process->istances_wait_numb++;

      g_mutex_unlock (data->mutex);

      log_write_generic (data, LOG_VERBOSE_INFO, LOG_ISTANCE_WAIT,
			 "process", LOG_VERBOSE_INFO, LOG_TYPE_STRING,
			 process->id, NULL);
      return;
    }

  /* Must I use the input buffer or the parent input? */
  if (process->input_from_parent == FALSE)
    input_buffer = process->input;

  /* Trim the input: */
  if (input_buffer && process->input_trim == TRUE
      && !(input_buffer = io_string_trim (input_buffer, &error)))
    {
      log_write_generic (data, LOG_VERBOSE_ERROR, LOG_ISTANCE_ERROR,
			 "reason", LOG_VERBOSE_ERROR, LOG_TYPE_STRING,
			 error->message,
			 "command", LOG_VERBOSE_ERROR, LOG_TYPE_STRING,
			 process->command, NULL);

      g_error_free (error);
      return;
    }

  /* Split the command: */
  argv = istance_split (process->command);

  /* Inline attribute: */
  if (process->command_inline == TRUE)
    {
      tmp = istance_command_inline (argv, input_buffer);
      g_strfreev (argv);
      argv = tmp;
    }

  /* Variables: */
  if ((tmp = istance_env_command (argv, data->envs)))
    {
      g_strfreev (argv);
      argv = tmp;
    }

  /* Executing the process: */
  if (
#ifndef G_OS_UNIX
       g_spawn_async_with_pipes (NULL, argv, NULL,
				 G_SPAWN_DO_NOT_REAP_CHILD |
				 G_SPAWN_SEARCH_PATH,
				 (GSpawnChildSetupFunc) istance_env,
				 data->envs, &pid, &in, &out, &err, &error)
#else
       istance_spawn (argv, (GSpawnChildSetupFunc) istance_env, data->envs,
		      &pid, &in, &out, &err, &error)
#endif
       == FALSE)

    {
      log_write_generic (data, LOG_VERBOSE_ERROR, LOG_ISTANCE_ERROR,
			 "reason", LOG_VERBOSE_ERROR, LOG_TYPE_STRING,
			 error->message,
			 "command", LOG_VERBOSE_ERROR, LOG_TYPE_STRING,
			 process->command, NULL);

      io_string_unref (input_buffer);
      g_error_free (error);
      g_strfreev (argv);
      return;
    }

  g_strfreev (argv);

  /* Create the data struct for the new istance: */
  istance = g_malloc0 (sizeof (TBIstance));
  istance_ref (istance);

  if (!(istance->output = io_string_new_empty (&data->swap, &error)))
    {
      log_write_generic (data, LOG_VERBOSE_ERROR, LOG_ISTANCE_ERROR,
			 "reason", LOG_VERBOSE_ERROR, LOG_TYPE_STRING,
			 error->message,
			 "command", LOG_VERBOSE_ERROR, LOG_TYPE_STRING,
			 process->command, NULL);

      io_string_unref (input_buffer);
      istance_free (istance);
      g_error_free (error);
      return;
    }

  /* This past has to be locked: */
  g_mutex_lock (data->mutex);

  data->istances = g_list_prepend (data->istances, istance);

  istance->data = data;
  istance->pid = pid;

  istance->process = process;
  istance->process_order = order;

  process->istances_running++;

  /* Duplicate the metaprocess for this istance: */
  if (metaprocesses)
    {
      istance->metaprocesses = g_list_copy (metaprocesses);
      g_list_foreach (istance->metaprocesses, (GFunc) process_ref, NULL);
    }

  /* Setting the parent of this istance: */
  istance->parents = g_list_copy (parents);
  g_list_foreach (istance->parents, (GFunc) istance_ref, NULL);

  /* Input configuration: */
  istance->input = input_buffer;

  if (process->input_trim == FALSE)
    io_string_ref (input_buffer);

  g_mutex_unlock (data->mutex);

  /* Time: */
  istance->time = process->last_scheduled = time (NULL);

#ifdef G_OS_WIN32
  istance->io_input = g_io_channel_win32_new_fd (in);
#else
  istance->io_input = g_io_channel_unix_new (in);
#endif

  g_io_channel_set_encoding (istance->io_input, NULL, NULL);
  g_io_channel_set_flags (istance->io_input,
			  g_io_channel_get_flags (istance->
						  io_input) |
			  G_IO_FLAG_NONBLOCK, NULL);

  /* The watch function for the input, output and error: */
  if (process->command_inline == FALSE && istance->input
      && io_string_get_size (istance->input) > 0)
    {
      istance->io_input_source =
	g_io_add_watch (istance->io_input, G_IO_OUT, (GIOFunc) istance_io_out,
			istance);
    }

  /* Close: */
  else
    {
      g_io_channel_shutdown (istance->io_input, TRUE, NULL);
      g_io_channel_unref (istance->io_input);
      istance->io_input = NULL;
    }

#ifdef G_OS_WIN32
  istance->io_output = g_io_channel_win32_new_fd (out);
  istance->io_error = g_io_channel_win32_new_fd (err);
#else
  istance->io_output = g_io_channel_unix_new (out);
  istance->io_error = g_io_channel_unix_new (err);
#endif

  g_io_channel_set_encoding (istance->io_output, NULL, NULL);
  g_io_channel_set_flags (istance->io_output,
			  g_io_channel_get_flags (istance->
						  io_output) |
			  G_IO_FLAG_NONBLOCK, NULL);

  istance->io_output_source =
    g_io_add_watch (istance->io_output, G_IO_IN, (GIOFunc) istance_io_in,
		    istance);

  g_io_channel_set_encoding (istance->io_error, NULL, NULL);
  g_io_channel_set_flags (istance->io_error,
			  g_io_channel_get_flags (istance->
						  io_error) |
			  G_IO_FLAG_NONBLOCK, NULL);

  istance->io_error_source =
    g_io_add_watch (istance->io_error, G_IO_IN, (GIOFunc) istance_io_in_err,
		    istance);

#ifndef G_OS_UNIX
  g_child_watch_add (istance->pid, (GChildWatchFunc) istance_exit, istance);
#endif

  log_write_istance (data, istance, LOG_VERBOSE_INFO, LOG_ISTANCE_BORN,
		     "input", LOG_VERBOSE_DEBUG, LOG_TYPE_IOSTRING,
		     istance->input, "inline", LOG_VERBOSE_INFO,
		     LOG_TYPE_BOOLEAN, process->command_inline, "parent",
		     LOG_VERBOSE_INFO, LOG_TYPE_INTEGER, parents ?
		     ((TBIstance *) (parents->data))->pid : 0, "trim",
		     LOG_VERBOSE_INFO, LOG_TYPE_BOOLEAN, process->input_trim,
		     "fromParent", LOG_VERBOSE_DEBUG, LOG_TYPE_BOOLEAN,
		     process->input_from_parent, NULL);
}

/* This function creates a new istance from a process: */
void
istance_new_from_process (TBGlobal * data, TBProcess * process)
{
#ifdef TB_DEBUG
  debug (__func__, __LINE__, "process: %s", process->id);
#endif
  istance_new_exec (data, process, 0, NULL, NULL, process->input, FALSE);
}

/* This function creates a new istance from a meta process: */
void
istance_new_from_meta_process (TBGlobal * data, TBProcess * process,
			       GList * metaprocesses, TBIstance * istance)
{
#ifdef TB_DEBUG
  debug (__func__, __LINE__, "process: %s, metaprocesses: %p, istance %p",
	 process->id, metaprocesses, istance);
#endif
  istance_new_exec (data, process, 0, istance->parents, metaprocesses,
		    istance->input, FALSE);
}

/* This function creates a new istance from another istance: */
void
istance_new_from_istance (TBGlobal * data, TBIstance * parent,
			  TBProcess * process, guint order,
			  TBIOString * input)
{
#ifdef TB_DEBUG
  debug (__func__, __LINE__, "istance: %p, process: %s, order: %d, input: %p",
	 parent, process->id, order, input);
#endif

  parent->parents = g_list_prepend (parent->parents, parent);

  istance_new_exec (data, process, order, parent->parents,
		    parent->metaprocesses, input, FALSE);

  parent->parents = g_list_remove (parent->parents, parent);
}

/* This function creates a new istance from the wait list: */
gboolean
istance_new_from_wait (TBGlobal * data, TBIstanceWait * wait)
{
#ifdef TB_DEBUG
  debug (__func__, __LINE__, "wait: %p", wait);
#endif

  /* If it can be execute: */
  if (wait->process->istances <= wait->process->istances_running)
    return FALSE;

  wait->process->istances_wait_numb--;
  data->istances_wait_numb--;

  data->istances_wait = g_list_remove (data->istances_wait, wait);

  istance_new_exec (data, wait->process, wait->process_order, wait->parents,
		    wait->metaprocesses, wait->input, TRUE);

  g_list_foreach (wait->parents, (GFunc) istance_unref, NULL);
  istance_wait_free (wait);

  return TRUE;
}

/* This function split a command into an array: */
static gchar **
istance_split (gchar * ptr)
{
  gboolean q1, q2, q3;
  GString *opt;
  gunichar unichr;
  GList *ret = NULL;
  GList *list;
  gchar **array;
  gint i;

  opt = g_string_new (NULL);
  q1 = q2 = q3 = FALSE;

  while ((unichr = g_utf8_get_char (ptr)))
    {
      ptr = g_utf8_next_char (ptr);

      if (q3 == TRUE)
	{
	  opt = g_string_append_unichar (opt, unichr);
	  q3 = FALSE;
	}

      else if (unichr == '\\')
	{
	  opt = g_string_append_unichar (opt, unichr);
	  q3 = TRUE;
	}

      else if (unichr == '\"' && q2 == FALSE)
	q1 = !q1;

      else if (unichr == '\'' && q1 == FALSE)
	q2 = !q2;

      else if ((unichr == ' ' || unichr == '\t') && q1 == FALSE
	       && q2 == FALSE)
	{
	  if (opt->len)
	    {
	      ret = g_list_append (ret, g_string_free (opt, FALSE));
	      opt = g_string_new (NULL);
	    }
	}

      else
	opt = g_string_append_unichar (opt, unichr);
    }

  if (opt->len)
    ret = g_list_append (ret, g_string_free (opt, FALSE));
  else
    g_string_free (opt, TRUE);

  array = g_malloc (sizeof (gchar *) * (g_list_length (ret) + 1));
  for (i = 0, list = ret; list; list = list->next)
    array[i++] = list->data;

  array[i] = NULL;
  g_list_free (ret);
  return array;
}

/* This function resolves the inline request: */
static gchar **
istance_command_inline (gchar ** argv, TBIOString * input)
{
  gchar **ret;
  gint i, max;
  gboolean ok = FALSE;

  for (max = 0; argv && argv[max]; max++);

  ret = g_malloc (sizeof (gchar *) * (max + 2));

  for (i = 0; argv && argv[i]; i++)
    {
      if (input && !strcmp (argv[i], "%s"))
	{
	  /* FIXME: This function returns only part of the string: */
	  ret[i] = io_string_to_gchar (input, 0, NULL);
	  ok = TRUE;
	}

      else
	ret[i] = g_strdup (argv[i]);
    }

  if (input && ok == FALSE)
    {
      /* FIXME: This function returns only part of the string: */
      ret[i] = io_string_to_gchar (input, 0, NULL);
      i++;
    }

  ret[i] = NULL;
  return ret;
}

/* This function is called when a process exits */
void
istance_exit (GPid pid, gint status, TBIstance * istance)
{
  GList *list;

#ifdef TB_DEBUG
  debug (__func__, __LINE__, "pid: %d, status: %d, istance: %p", pid, status,
	 istance);
#endif

  /* Remove everything from the output iochannel: */
  if (istance->io_output_source)
    {
      g_source_remove (istance->io_output_source);
      istance->io_output_source = 0;

      while (istance_io_in (istance->io_output, 0, istance) == TRUE);
    }

  /* Remove everything from the error iochannel: */
  if (istance->io_error_source)
    {
      g_source_remove (istance->io_error_source);
      istance->io_error_source = 0;

      while (istance_io_in_err (istance->io_error, 0, istance) == TRUE);
    }

  /* Closing the iochannels (and the filedescriptors): */
  istance_io_close (istance, ISTANCE_IO_IN);
  istance_io_close (istance, ISTANCE_IO_OUT);
  istance_io_close (istance, ISTANCE_IO_ERR);

  /* Errors ? */
  if (istance->error)
    log_write_istance (istance->data, istance, LOG_VERBOSE_ERROR,
		       LOG_ISTANCE_ERROR, "error_message", LOG_VERBOSE_ERROR,
		       LOG_TYPE_IOSTRING, istance->error, NULL);

  /* Log the status: */
  log_write_istance (istance->data, istance, LOG_VERBOSE_INFO,
		     LOG_ISTANCE_DIED, "status", LOG_VERBOSE_INFO,
		     LOG_TYPE_INTEGER, status, "output", LOG_VERBOSE_DEBUG,
		     LOG_TYPE_IOSTRING, istance->output, NULL);

  g_mutex_lock (istance->data->mutex);
  istance->process->istances_running--;
  g_mutex_unlock (istance->data->mutex);

  /* Propagation: */
  if (status == 0)
    istance_propage_check (istance, istance->process, 0);

  g_mutex_lock (istance->data->mutex);

  /* Removing from the list of istances: */
  istance->data->istances = g_list_remove (istance->data->istances, istance);

  g_mutex_unlock (istance->data->mutex);

  /* Unref: */
  for (list = istance->parents; list; list = list->next)
    {
      TBIstance *old = list->data;

      /* The last ref: */
      if (old->ref == 1)
	istance_propage_check (old, old->process, old->process_order + 1);

      istance_unref (old);
    }

  istance_unref (istance);
}

/* This function writes a log message and kills the istance if a I/O
 * operation fails: */
static void
istance_io_error (TBIstance * istance, gchar * operation, GError * error)
{
  log_write_istance (istance->data, istance, LOG_VERBOSE_ERROR,
		     LOG_ISTANCE_ERROR,
		     "operation", LOG_VERBOSE_ERROR, LOG_TYPE_STRING,
		     operation,
		     "error", LOG_VERBOSE_ERROR, LOG_TYPE_STRING,
		     error->message, NULL);
  g_error_free (error);

#ifdef G_OS_UNIX
  kill (istance->pid, SIGQUIT);
#endif
}

/* This function is called when there is something to read */
static gboolean
istance_io_in (GIOChannel * source, GIOCondition condition,
	       TBIstance * istance)
{
  GError *error = NULL;
  gchar buffer[1024];
  GIOStatus status;
  gsize read;

  status =
    g_io_channel_read_chars (source, buffer, sizeof (buffer), &read, &error);

  if (status == G_IO_STATUS_ERROR)
    {
      /* Stop the output process: */
      istance_io_error (istance, "read", error);
      istance_io_close (istance, ISTANCE_IO_OUT);
      return FALSE;
    }

  /* Stop the output process: */
  if (status == G_IO_STATUS_EOF)
    {
      istance_io_close (istance, ISTANCE_IO_OUT);
      return FALSE;
    }

  /* Good: */
  if (status == G_IO_STATUS_NORMAL)
    {
      if (io_string_append (istance->output, buffer, read, &error) == FALSE)
	{
	  /* Stop the output process: */
	  istance_io_error (istance, "read", error);
	  istance_io_close (istance, ISTANCE_IO_OUT);
	  return FALSE;
	}
    }

  return TRUE;
}

/* This function is called when there is something to read */
static gboolean
istance_io_in_err (GIOChannel * source, GIOCondition condition,
		   TBIstance * istance)
{
  GError *error = NULL;
  gchar buffer[1024];
  GIOStatus status;
  gsize read;

  status =
    g_io_channel_read_chars (source, buffer, sizeof (buffer), &read, &error);

  if (status == G_IO_STATUS_ERROR)
    {
      /* Stop the error process: */
      istance_io_error (istance, "read", error);
      istance_io_close (istance, ISTANCE_IO_ERR);
      return FALSE;
    }

  /* Stop the error process: */
  if (status == G_IO_STATUS_EOF)
    {
      istance_io_close (istance, ISTANCE_IO_ERR);
      return FALSE;
    }

  /* Good: */
  if (status == G_IO_STATUS_NORMAL)
    {
      if (!istance->error
	  && !(istance->error =
	       io_string_new_empty (&istance->data->swap, &error)))
	{
	  istance_io_error (istance, "read", error);
	  istance_io_close (istance, ISTANCE_IO_ERR);
	  return FALSE;
	}

      if (io_string_append (istance->error, buffer, read, &error) == FALSE)
	{
	  istance_io_error (istance, "read", error);
	  istance_io_close (istance, ISTANCE_IO_ERR);
	  return FALSE;
	}
    }

  return TRUE;
}

/* This function is called when there is something to write */
static gboolean
istance_io_out (GIOChannel * source, GIOCondition condition,
		TBIstance * istance)
{
  GError *error = NULL;
  GIOStatus status;
  gchar *string;
  gsize size;
  gsize ret;

  /* Try to write something: */
  if (!
      (string =
       io_string_to_gchar (istance->input, istance->input_written, &size)))
    return TRUE;

  if ((status =
       g_io_channel_write_chars (source, string, size, &ret,
				 &error)) == G_IO_STATUS_NORMAL)
    status = g_io_channel_flush (source, &error);

  g_free (string);

  /* If there is an error, kill th process: */
  if (status == G_IO_STATUS_ERROR)
    {
      /* Close the channel and stop the input process: */
      istance_io_error (istance, "write", error);
      istance_io_close (istance, ISTANCE_IO_IN);
      return FALSE;
    }

  /* If it is fine: */
  if (status == G_IO_STATUS_NORMAL)
    {
      istance->input_written += ret;

      /* Everything written? */
      if (istance->input_written < io_string_get_size (istance->input))
	return TRUE;

      /* If yes, close the input channel: */
      istance_io_close (istance, ISTANCE_IO_IN);
      return FALSE;
    }

  return TRUE;
}

/* PROPAGE ******************************************************************/
typedef struct tb_istance_propage_t TBIstancePropage;
struct tb_istance_propage_t
{
  TBIstance *istance;
  guint order;
};

static void istance_propage (TBIOString * output, TBIstancePropage * propage);
static void istance_propage_meta (TBIOString * output,
				  TBIstancePropage * propage);

static void
istance_propage_check (TBIstance * istance, TBProcess * process, guint order)
{
  TBIstancePropage propage;

  TBIOStringSplitFunc func;
  GError *error = NULL;

#ifdef TB_DEBUG
  debug (__func__, __LINE__, "istance: %p, process: %s, order: %d", istance,
	 process->id, order);
#endif

  if (istance_is_to_propage (istance, order) == FALSE)
    return;

  propage.istance = istance;
  propage.order = order;

  if (istance->process->meta == TRUE)
    func = (TBIOStringSplitFunc) istance_propage_meta;
  else
    func = (TBIOStringSplitFunc) istance_propage;

  if (istance->process->multiple == FALSE)
    {
      func (istance->output, &propage);
    }

  else
    if (io_string_split
	(istance->output, TB_ISTANCE_SPLITTER, func, &propage,
	 &error) == FALSE)
    {
      log_write_istance (istance->data, istance, LOG_VERBOSE_ERROR,
			 LOG_ISTANCE_ERROR, "error_message",
			 LOG_VERBOSE_ERROR, LOG_TYPE_IOSTRING, error->message,
			 NULL);

      g_error_free (error);
    }
}

static gboolean
istance_is_to_propage (TBIstance * istance, guint order)
{
  GList *list;

  if (!istance || !istance->process)
    return FALSE;

  if (istance->process->meta == TRUE)
    return TRUE;

  for (list = istance->process->processes; list; list = list->next)
    {
      TBSubProcess *sub = list->data;

      if (sub->order >= order)
	return TRUE;
    }

  return FALSE;
}

static void
istance_propage (TBIOString * output, TBIstancePropage * propage)
{
  GList *list;
  guint min = 0;
  gboolean min_first = TRUE;

#ifdef TB_DEBUG
  debug (__func__, __LINE__, "propage: %p", propage);
#endif
  istance_check_wait (propage->istance->data);

  /* Looking for the min order value: */
  for (list = propage->istance->process->processes; list; list = list->next)
    {
      TBSubProcess *sub = list->data;

      if (sub->order >= propage->order)
	{
	  if (min_first == TRUE)
	    {
	      min = sub->order;
	      min_first = FALSE;
	    }

	  else if (sub->order < min)
	    min = sub->order;
	}
    }

  if (min_first == TRUE)
    return;

  /* Set the new level of order: */
  propage->istance->process_order = min;

  /* For any children... */
  for (list = propage->istance->process->processes; list; list = list->next)
    {
      GList *l;
      TBSubProcess *sub = list->data;

      if (sub->order != min)
	continue;

      /* it search the correnct process: */
      for (l = propage->istance->metaprocesses; l; l = l->next)
	{
	  TBProcess *process = l->data;

	  /* Bingo! */
	  if (!g_utf8_collate (process->id, sub->process))
	    {
	      /* Propagate: */
	      istance_new_from_istance (propage->istance->data,
					propage->istance, process, min,
					output);

	      break;
	    }
	}

      /* Founded in the metaprocess: */
      if (l)
	continue;

      /* it search the current process: */
      for (l = propage->istance->data->processes; l; l = l->next)
	{
	  TBProcess *process = l->data;

	  /* Bingo! */
	  if (!g_utf8_collate (process->id, sub->process))
	    {
	      /* Propagate: */
	      istance_new_from_istance (propage->istance->data,
					propage->istance, process, min,
					output);

	      break;
	    }
	}

      /* Founded in the processes! */
      if (l)
	continue;

      log_write_istance (propage->istance->data, propage->istance,
			 LOG_VERBOSE_ERROR, LOG_ISTANCE_ERROR,
			 "process_error", LOG_VERBOSE_ERROR, LOG_TYPE_STRING,
			 sub->process, "error", LOG_VERBOSE_ERROR,
			 LOG_TYPE_STRING, "This process doesn't exist", NULL);
    }
}

static void
istance_propage_meta (TBIOString * output, TBIstancePropage * propage)
{
  GList *l;

  nxml_t *nxml;
  nxml_data_t *cur;

  gsize size;
  guint64 offset;

  GList *processes;
  GList *newprocesses;
  GError *error = NULL;

  GString *string;
  gchar *tmp;

#ifdef TB_DEBUG
  debug (__func__, __LINE__, "propage: %p", propage);
#endif

  istance_check_wait (propage->istance->data);

  if (nxml_new (&nxml) != NXML_OK)
    {
      log_write_istance (propage->istance->data, propage->istance,
			 LOG_VERBOSE_ERROR, LOG_ISTANCE_ERROR, "parser_error",
			 LOG_VERBOSE_ERROR, LOG_TYPE_STRING,
			 "initialization error", NULL);

      return;
    }

  string = g_string_new (NULL);
  offset = 0;

  while ((tmp = io_string_to_gchar (output, offset, &size)))
    {
      offset += size;

      string = g_string_append_len (string, tmp, size);
      g_free (tmp);
    }

  tmp = g_string_free (string, FALSE);

  if (nxml_parse_buffer (nxml, tmp, offset) != NXML_OK
      || nxml_root_element (nxml, &cur) != NXML_OK
      || strcmp (cur->value, TB_CONTROLLER_ROOT_TAG))
    {
      log_write_istance (propage->istance->data, propage->istance,
			 LOG_VERBOSE_ERROR, LOG_ISTANCE_ERROR, "parser_error",
			 LOG_VERBOSE_ERROR, LOG_TYPE_STRING,
			 "XML format not recognized", "parser_input",
			 LOG_VERBOSE_DEBUG, LOG_TYPE_IOSTRING, output, NULL);

      nxml_free (nxml);
      g_free (tmp);
      return;
    }

  g_free (tmp);

  if (process_parser (propage->istance->data, nxml, &processes, TRUE, &error)
      == FALSE)
    {
      log_write_istance (propage->istance->data, propage->istance,
			 LOG_VERBOSE_ERROR, LOG_ISTANCE_ERROR, "parser_error",
			 LOG_VERBOSE_ERROR, LOG_TYPE_STRING, error->message,
			 "parser_input", LOG_VERBOSE_DEBUG, LOG_TYPE_IOSTRING,
			 output, NULL);

      nxml_free (nxml);
      return;
    }

  nxml_free (nxml);

  if (!processes)
    return;

  /* Create a new list with the new processes and the old metaprocesses
   * of the current istance: */
  newprocesses = g_list_copy (processes);

  if (propage->istance->metaprocesses)
    {
      GList *tmp = g_list_copy (propage->istance->metaprocesses);
      newprocesses = g_list_concat (newprocesses, tmp);
    }

  for (l = processes; l; l = l->next)
    {
      TBProcess *process = l->data;

      if (process->global == TRUE)
	istance_new_from_meta_process (propage->istance->data, process,
				       newprocesses, propage->istance);
    }

  g_list_free (newprocesses);
  g_list_free (processes);
}

/* ENV **********************************************************************/
static void
istance_env (GList * list)
{
  TBEnv *env;

  for (; list; list = list->next)
    {
      env = list->data;

      g_setenv (env->env, env->value, TRUE);
    }

  g_setenv ("TBCONTROLLER_SPLITTER", TB_ISTANCE_SPLITTER, TRUE);
}

static gchar **
istance_env_command (gchar ** argv, GList * list)
{
  gint i;
  gchar **ret;

  for (i = 0; argv && argv[i]; i++)
    if (g_str_has_prefix (argv[i], "$") == TRUE)
      break;

  if (argv[i] == NULL)
    return NULL;

  /* Count the last one: */
  for (; argv && argv[i]; i++);

  ret = g_malloc (sizeof (gchar *) * (i + 1));

  for (i = 0; argv && argv[i]; i++)
    {
      if (g_str_has_prefix (argv[i], "$") == FALSE)
	ret[i] = g_strdup (argv[i]);

      else
	{
	  GList *l;

	  for (l = list; l; l = l->next)
	    {
	      TBEnv *env = l->data;

	      if (!strcmp (env->env, argv[i] + 1))
		{
		  ret[i] = g_strdup (env->value);
		  break;
		}
	    }

	  if (!l)
	    ret[i] = g_strdup (argv[i]);
	}
    }

  ret[i] = NULL;
  return ret;
}

/* WAIT CHECK ***************************************************************/
static void
istance_check_wait (TBGlobal * data)
{
#ifdef TB_DEBUG
  debug (__func__, __LINE__, NULL);
#endif

  while (data->istances_wait_numb > data->wait_max_istance)
    {
      while (g_main_context_pending (NULL))
	g_main_context_iteration (NULL, TRUE);

      g_usleep (100);
    }
}

/* REF and FREE *************************************************************/
void
istance_ref (TBIstance * istance)
{
  if (!istance)
    return;

  istance->ref++;
}

void
istance_unref (TBIstance * istance)
{
  if (!istance)
    return;

  istance->ref--;

  if (istance->ref <= 0)
    istance_free (istance);
}

void
istance_free (TBIstance * istance)
{
  if (!istance)
    return;

  if (istance->metaprocesses)
    {
      g_list_foreach (istance->metaprocesses, (GFunc) process_unref, NULL);
      g_list_free (istance->metaprocesses);
    }

  if (istance->parents)
    g_list_free (istance->parents);

  if (istance->pid)
    g_spawn_close_pid (istance->pid);

  istance_io_close (istance, ISTANCE_IO_IN);
  istance_io_close (istance, ISTANCE_IO_OUT);
  istance_io_close (istance, ISTANCE_IO_ERR);

  if (istance->input)
    io_string_unref (istance->input);

  if (istance->output)
    io_string_unref (istance->output);

  if (istance->error)
    io_string_unref (istance->error);

  g_free (istance);
}

static void
istance_io_close (TBIstance * istance, TBIstanceIO io)
{
  if (!istance)
    return;

  switch (io)
    {
    case ISTANCE_IO_IN:
      if (istance->io_input_source)
	{
	  g_source_remove (istance->io_input_source);
	  istance->io_input_source = 0;
	}

      if (istance->io_input)
	{
	  g_io_channel_shutdown (istance->io_input, TRUE, NULL);
	  g_io_channel_unref (istance->io_input);
	  istance->io_input = NULL;
	}

      break;

    case ISTANCE_IO_OUT:
      if (istance->io_output_source)
	{
	  g_source_remove (istance->io_output_source);
	  istance->io_output_source = 0;
	}

      if (istance->io_output)
	{
	  g_io_channel_shutdown (istance->io_output, TRUE, NULL);
	  g_io_channel_unref (istance->io_output);
	  istance->io_output = NULL;
	}

      break;

    case ISTANCE_IO_ERR:
      if (istance->io_error_source)
	{
	  g_source_remove (istance->io_error_source);
	  istance->io_error_source = 0;
	}

      if (istance->io_error)
	{
	  g_io_channel_shutdown (istance->io_error, TRUE, NULL);
	  g_io_channel_unref (istance->io_error);
	  istance->io_error = NULL;
	}
      break;


    default:
      break;
    }
}

static void
istance_wait_free (TBIstanceWait * wait)
{
  if (!wait)
    return;

  if (wait->metaprocesses)
    {
      g_list_foreach (wait->metaprocesses, (GFunc) process_unref, NULL);
      g_list_free (wait->metaprocesses);
    }

  if (wait->parents)
    g_list_free (wait->parents);

  if (wait->input)
    io_string_unref (wait->input);

  g_free (wait);
}

#ifdef G_OS_UNIX
/* SPAWN ********************************************************************/
static gboolean
istance_spawn (gchar ** argv, GSpawnChildSetupFunc child_setup,
	       gpointer user_data, GPid * ret_pid, gint * standard_input,
	       gint * standard_output, gint * standard_error, GError ** error)
{
  GPid pid;
  gint fdinput[2];
  gint fdoutput[2];
  gint fderror[2];

  if (pipe (fdinput) == -1)
    {
      g_set_error (error, controller_error_quark (), 0,
		   "Error creating a pipe: %s", g_strerror (errno));
      return FALSE;
    }

  if (pipe (fdoutput) == -1)
    {
      close (fdinput[0]);
      close (fdinput[1]);

      g_set_error (error, controller_error_quark (), 0,
		   "Error creating a pipe: %s", g_strerror (errno));
      return FALSE;
    }

  if (pipe (fderror) == -1)
    {
      close (fdinput[0]);
      close (fdinput[1]);

      close (fdoutput[0]);
      close (fdoutput[1]);

      g_set_error (error, controller_error_quark (), 0,
		   "Error creating a pipe: %s", g_strerror (errno));
      return FALSE;
    }

  if ((pid = fork ()) == -1)
    {
      close (fdinput[0]);
      close (fdinput[1]);

      close (fdoutput[0]);
      close (fdoutput[1]);

      close (fderror[0]);
      close (fderror[1]);

      g_set_error (error, controller_error_quark (), 0, "Error forking: %s",
		   g_strerror (errno));
      return FALSE;
    }

  /* Child: */
  if (pid == 0)
    {
      close (fdinput[1]);
      dup2 (fdinput[0], STDIN_FILENO);

      close (fdoutput[0]);
      dup2 (fdoutput[1], STDOUT_FILENO);

      close (fderror[0]);
      dup2 (fderror[1], STDERR_FILENO);

      if (child_setup)
	child_setup (user_data);

      if (execvp (argv[0], argv) == -1)
	{
	  g_fprintf (stderr, "Error executing '%s': %s", argv[0],
		     g_strerror (errno));

	  close (fdinput[0]);
	  close (fdoutput[1]);
	  close (fderror[1]);

	  exit (EXIT_FAILURE);
	}
    }

  /* Father: */
  close (fdinput[0]);
  close (fdoutput[1]);
  close (fderror[1]);

  *ret_pid = pid;

  *standard_input = fdinput[1];
  *standard_output = fdoutput[0];
  *standard_error = fderror[0];

  return TRUE;
}
#endif

/* EOF */
